

namespace Stee.CAP8.MessageBus.Kafka.Subscriber
{
    using System;
    using System.Threading;
    using System.Threading.Tasks;
    using NLog;
    using Stee.CAP8.MessageBus.Interface.Delegates;
    using Stee.CAP8.MessageBus.Interface.Interfaces;
    using Stee.CAP8.MessageBus.Interface.Models;

    /* 
     * COMMAND Messages
     * =====
     * Individial services can be designated with a group ID solely for command type messsages.
     * For notification service, it could be "notificationsvc", where messages to this topic will be consumed
     * only by notification service. 
     * Example command group IDs:
	 *     - notificationsvc
	 *     - dispatchsvc
	 *     - gissvc
     *
     *
     * EVENT Messages
     * =====
     * Based on events, if 2 services share the same group ID, only 1 service will process a specific message.
     * This is not ideal if the event was "unitlocation_update" where resource and GIS services require this message.
     * Therefore to allow both services access to an event message, 2 group IDs must be created.
     * Example event group IDs:
     *     - unit_create_{0}: notificationsvc, resourcesvc
     *     - unit_update_{0}: notificationsvc, resourcesvc
     *     - Eg.: unit_update_notificationsvc
     *
     *     - incident_create_{0}: notificationsvc, resourcesvc
     *     - incident_update_{0}: notificationsvc, resourcesvc
     *     - Eg.: incident_update_notificationsvc
     *
     *     - unitlocation_update_{0}: notificationsvc, dispatchsvc, gissvc
     *     - incidentlocation_update_{0}: notificationsvc, dispatchsvc, gissvc
     *     - Eg.: incidentlocation_update_notificationsvc
     *
     * Thereafter, command & event messages can be viewed as similar implementation.
     * 
     *
     * CONSUMER
     * =====
     * Kafka consumer uses polling to 'listen' to messages, therefore is not automatically made known to new messages.
     * Using a polling method, the consumer has a while loop to constantly poll for new messages(never ending).
     * On subscription, a polling task has to be created and maintained preferrably by the subscriber themselves.
     * The polling task will have a timer to check kafka consumer state.
     * If the consumer is 'stopped', the timer will trigger an event to restart message polling
     * 
     * Required information: (Topic, notification handler). Configuration is assumed to be known.
     * Logic: A recoverable task constantly polling for messages with subscribe as start, unsubscribe as stop.
     * 
     * 
     * SERIALIZATION / DESERIALIZATION Issue
     * =====
     * Kafka is able to publish messages of any serializable data type. The serialized message in byte[] will be published to
     * the broker. Since there is no limitation to the number of allowed data types per Kafka topic, a topic can potentially
     * contain messages of ANY data type. As a publisher, this will not be an issue as the type is known during
     * publishing (serialization).
     * 
     * As a consumer, a logic is needed to handle multiple data types even if it consumes only 1 topic. To complicate
     * things, the consumed message will need de-serialization which requires the data type information.
     * The solution to this is to save the data type with the published message. During consumption, the data type can be
     * extracted and with proper mapping (logic), de-serialized.
     * In the publisher implementation, data type will be saved as the last message header with constant key, "MessageType".
     * Consumer will then use the same key to retrieve header for the data type.
     * https://github.com/confluentinc/confluent-kafka-dotnet/issues/1183
     * 
     */
    /// <summary>
    /// Implements the subscriber/consumer for receiving messages. 
    /// Each subscriber will handle a specific messsage data type. 
    /// https://github.com/confluentinc/confluent-kafka-dotnet/blob/master/examples/Consumer/Program.cs
    /// </summary>
    public class SubscriberImpl : ISubscriber, IDisposable
    {
        #region Constant & Static Variables

        private readonly static Logger Logger = LogManager.GetCurrentClassLogger();

        #endregion Constant & Static Variables


        #region Fields

        /// <summary>
        /// Polling implmentation
        /// </summary>
        private PollingImpl pollingImpl;

        /// <summary>
        /// Asynchronous operation for polling consumer messages
        /// </summary>
        private Task pollingTask;

        /// <summary>
        /// Cancel token for polling task
        /// </summary>
        private CancellationTokenSource cancellationToken;

        #endregion Fields


        #region Properties

        /// <summary>
        /// Gets the message topic
        /// Mandatory configuration.
        /// </summary>
        public TopicName TopicName
        {
            get { return this.pollingImpl.TopicName; }
        }

        /// <summary>
        /// Gets the client group id string.
        /// All clients sharing the same group.id belong to the same group.
        /// Mandatory configuration.
        /// </summary>
        public string GroupID
        {
            get { return this.pollingImpl?.GroupID; }
        }

        /// <summary>
        /// Gets an optional identifier of a Kafka consumer(in a consumer group) that is passed to a Kafka broker with every request.
        /// The sole purpose of this is to be able to track the source of requests beyond just ip and port
        /// by allowing a logical application name to be included in Kafka logs and monitoring aggregates.
        /// </summary>
        public string ClientID
        {
            get { return this.pollingImpl?.ClientID; }
        }

        /// <summary>
        /// Gets the flag indicating that subscriber is initialized and ready for reading messages
        /// </summary>
        public bool IsInitialized { get; private set; } = false;

        /// <summary>
        /// Gets the flag indicating if consumer is currently consuming messages
        /// </summary>
        public bool IsConnected
        {
            get { return (this.pollingImpl == null) ? false : this.pollingImpl.IsConnected; }
        }

        /// <summary>
        /// Gets the flag indicating if publisher is disposed
        /// </summary>
        public bool IsDisposed
        {
            get { return (this.pollingImpl == null) ? true : this.pollingImpl.IsDisposed; }
        }

        #endregion Properties


        #region Constructors & Initializers

        /// <summary>
        /// Initializes a new instance of the SubscriberImpl class
        /// </summary>
        public SubscriberImpl() { }


        /// <summary>
        /// Initializes the subscription
        /// </summary>
        /// <param name="topicName">Subscription topic</param>
        /// <param name="subscribeTopicHandler">Subscribe message handler, message data type</param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        public bool Initialize(TopicName topicName, HandleSubscribeMessage subscribeTopicHandler, HandleBrokerError brokerErrorHandler)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Subscription to {topicName} is already initialised.");
                }

                this.pollingImpl = new PollingImpl(topicName, subscribeTopicHandler, brokerErrorHandler);
                this.IsInitialized = true;

                return this.IsInitialized;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Initializes the subscription
        /// </summary>
        /// <param name="topicName">Subscription topic</param>
        /// <param name="subscribedCommandHandler"></param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        public bool Initialize(TopicName topicName, HandleSubscribeCommand subscribedCommandHandler, HandleBrokerError brokerErrorHandler)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Subscription to {topicName} is already initialised.");
                }

                this.pollingImpl = new PollingImpl(topicName, subscribedCommandHandler, brokerErrorHandler);
                this.IsInitialized = true;

                return this.IsInitialized;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Initializes the subscription
        /// </summary>
        /// <param name="topicName">Subscription topic</param>
        /// <param name="subscribedEventHandler"></param>
        /// <param name="brokerErrorHandler">Broker error handler</param>
        /// <returns>Return TRUE if initialized else FALSE</returns>
        public bool Initialize(TopicName topicName, HandleSubscribeEvent subscribedEventHandler, HandleBrokerError brokerErrorHandler)
        {
            try
            {
                if (this.IsInitialized)
                {
                    throw new InvalidOperationException($"Subscription to {topicName} is already initialised.");
                }

                this.pollingImpl = new PollingImpl(topicName, subscribedEventHandler, brokerErrorHandler);
                this.IsInitialized = true;

                return this.IsInitialized;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        #endregion Constructors & Initializers


        #region Interface Implementation

        /// <summary>
        /// Subscribe to topic
        /// </summary>
        public void Subscribe()
        {
            try
            {
                this.cancellationToken = new CancellationTokenSource();
                this.pollingTask = Task.Run(() => this.pollingImpl.StartPolling(), this.cancellationToken.Token);
                this.pollingTask.ConfigureAwait(false);
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Commit offset of consumed message
        /// </summary>
        /// <param name="brokerMessage">Message consumed from Kafka cluster</param>
        public bool Commit(object brokerMessage)
        {
            return this.pollingImpl.Commit(brokerMessage);
        }

        /// <summary>
        /// Dispose subscriber
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        #endregion Interface Implementation


        #region Private Methods

        /// <summary>
        /// Cancel message subscription
        /// </summary>
        private void CancelTokenSource()
        {
            try
            {
                if (this.cancellationToken != null && !this.cancellationToken.IsCancellationRequested)
                {
                    this.cancellationToken.Cancel();
                    this.cancellationToken.Dispose();
                }

                this.cancellationToken = null;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Clear polling task
        /// </summary>
        private void ClearPollingTask()
        {
            try
            {
                //// Clear polling task
                if (this.pollingTask != null && (this.pollingTask.IsCanceled || this.pollingTask.IsCompleted || this.pollingTask.IsFaulted))
                {
                    this.pollingTask.Dispose();
                }

                this.pollingTask = null;
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        /// <summary>
        /// Unsubscribe from topic
        /// </summary>
        /// <returns>Return TRUE if unsubscribed successfully else FALSE</returns>
        private void Unsubscribe()
        {
            this.CancelTokenSource();
            this.ClearPollingTask();
        }

        /// <summary>
        /// Dispose
        /// </summary>
        /// <param name="disposing"></param>
        protected virtual void Dispose(bool disposing)
        {
            try
            {
                if (!this.IsDisposed)
                {
                    this.Unsubscribe();
                    this.pollingImpl.Dispose();
                }
            }
            catch (Exception ex)
            {
                Logger.Error(ex, ex.Message);
                throw;
            }
        }

        #endregion Private Methods
    }
}
